/*
 *    Copyright 2022 The DSMS Authors.
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

package com.dsms.modules.monitor.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.dsms.common.constant.*;
import com.dsms.common.exception.DsmsEngineException;
import com.dsms.common.model.Result;
import com.dsms.common.prometheus.builder.InstantQueryBuilder;
import com.dsms.common.prometheus.builder.QueryBuilderType;
import com.dsms.common.prometheus.builder.RangeQueryBuilder;
import com.dsms.common.prometheus.converter.result.DefaultQueryResult;
import com.dsms.common.prometheus.converter.result.MatrixData;
import com.dsms.common.prometheus.converter.result.QueryResultItemValue;
import com.dsms.common.prometheus.converter.result.VectorData;
import com.dsms.common.remotecall.model.FinishedDetail;
import com.dsms.common.remotecall.model.RemoteResponse;
import com.dsms.common.util.PrometheusUtils;
import com.dsms.dfsbroker.cluster.api.ClusterApi;
import com.dsms.dfsbroker.cluster.model.remote.StatusResult;
import com.dsms.dfsbroker.filesystem.FileSystem;
import com.dsms.dfsbroker.filesystem.service.IFileSystemService;
import com.dsms.dfsbroker.osd.osd.api.OsdApi;
import com.dsms.dfsbroker.osd.osd.model.remote.OSDStatusResult;
import com.dsms.dfsbroker.rbd.api.RbdApi;
import com.dsms.modules.monitor.model.dto.*;
import com.dsms.modules.monitor.model.vo.*;
import com.dsms.modules.monitor.service.IMonitorService;
import com.dsms.modules.util.PrometheusUtil;
import com.dsms.modules.util.RemoteCallUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.net.URI;
import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class MonitorServiceImpl implements IMonitorService {
    @Autowired
    ClusterApi clusterApi;

    @Autowired
    OsdApi osdApi;

    @Autowired
    RbdApi rbdApi;

    @Autowired
    IFileSystemService fileSystemService;

    @Override
    public Result<ClusterStatusDTO> listClusterStatusMon() {
        StatusResult clusterStatus;
        List<OSDStatusResult> osdStatusResults;
        try {
            clusterStatus = clusterApi.getStatus(RemoteCallUtil.generateRemoteRequest());
            osdStatusResults = osdApi.listOsdStatus(RemoteCallUtil.generateRemoteRequest());
        } catch (Throwable e) {
            throw new DsmsEngineException(ResultCode.MON_GETCLUSTERSTATUS_ERROR);
        }

        //1.set osd monitor
        //get osd which status both 'up' and 'in'
        List<OSDStatusResult> normalOsds = osdStatusResults.stream().filter(item -> item.getState().contains(OdsStatusEnum.EXISTS.getStatus()) && item.getState().contains(OdsStatusEnum.UP.getStatus()) && item.getWeight() > 0).collect(Collectors.toList());
        ClusterStatusDTO.OsdDTO osdDTO = new ClusterStatusDTO.OsdDTO(clusterStatus.getOsdmap().getNumOsds(), normalOsds.size());

        //2.set mon monitor
        List<String> nodeNames = osdStatusResults.stream().map(OSDStatusResult::getServer).distinct().collect(Collectors.toList());
        List<String> quorumNames = clusterStatus.getQuorumNames();
        List<String> downMonNames = nodeNames.stream().filter(item -> quorumNames.stream().noneMatch(item::equals)).collect(Collectors.toList());
        ClusterStatusDTO.MonDTO monDTO = new ClusterStatusDTO.MonDTO(clusterStatus.getMonmap().getNumMons(), quorumNames, downMonNames);

        //3.set mgr monitor
        //TODO mgr standby name
        ClusterStatusDTO.MgrDTO mgrDTO = new ClusterStatusDTO.MgrDTO(clusterStatus.getMgrmap().getLeader(), null, clusterStatus.getMgrmap().getNumStandbys());

        //4.set node mon
        //onlineNodes contains nodes with normal osd, mon, or mgr leader,a node is online as long as it meets one of these three conditions
        List<String> onlineNodes = new ArrayList<>();
        //5.get up ods node
        List<String> osdUpNodes = osdStatusResults.stream().filter(item -> item.getState().contains(OdsStatusEnum.EXISTS.getStatus()) && item.getState().contains(OdsStatusEnum.UP.getStatus())).map(OSDStatusResult::getServer).distinct().collect(Collectors.toList());
        onlineNodes.addAll(osdUpNodes);
        onlineNodes.addAll(quorumNames);
        onlineNodes.add(mgrDTO.getMgrLeader());
        //distinct array
        List<String> distinctOnlineNodes = onlineNodes.stream().filter(Objects::nonNull).distinct().collect(Collectors.toList());
        List<String> offlineNodes = nodeNames.stream().filter(item -> onlineNodes.stream().noneMatch(item::equals)).collect(Collectors.toList());
        ClusterStatusDTO.NodeDTO nodeDTO = new ClusterStatusDTO.NodeDTO(distinctOnlineNodes, offlineNodes);

        //6.set cpu usage
        InstantQueryBuilder instantQueryBuilder = PrometheusUtil.getQueryBuilder(QueryBuilderType.InstantQuery);
        URI cpuUsageUri = instantQueryBuilder.withQuery(PromQL.CLUSTER_CPU_USAGE).build();
        DefaultQueryResult<VectorData> cpuUsageResponse = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(cpuUsageUri, String.class));
        QueryResultItemValue cpuUsage = cpuUsageResponse.getQueryResult();

        //7.set memory usage
        URI memoryUsageUri = instantQueryBuilder.withQuery(PromQL.CLUSTER_MEMORY_USAGE).build();
        DefaultQueryResult<VectorData> memoryUsageResponse = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(memoryUsageUri, String.class));
        QueryResultItemValue memoryUsage = memoryUsageResponse.getQueryResult();

        //8.set capacity usage
        URI capacityUsageUri = instantQueryBuilder.withQuery(PromQL.CLUSTER_CAPACITY_USAGE).build();
        DefaultQueryResult<VectorData> capacityUsageResponse = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(capacityUsageUri, String.class));
        QueryResultItemValue capacityUsage = capacityUsageResponse.getQueryResult();

        //When there is no osd in the cluster, it returns null; when there is an osd, it returns a percentage value
        Double capacityUsagePercent = Optional.ofNullable(capacityUsage)
                .map(QueryResultItemValue::getPercentageValue)
                .orElse(null);

        //construct return object
        ClusterStatusDTO clusterStatusDTO = new ClusterStatusDTO(osdDTO, monDTO, mgrDTO, nodeDTO, cpuUsage.getPercentageValue(), memoryUsage.getPercentageValue(), capacityUsagePercent);

        return Result.OK(clusterStatusDTO);
    }

    @Override
    public Result<ClusterMonitorDTO> getClusterPerformanceMon(DurationVO durationVO) {
        long startTimeStamp = durationVO.getStartTimeStamp();
        long endTimeStamp = durationVO.getEndTimeStamp();
        String step = durationVO.getStep();

        RangeQueryBuilder rangeQueryBuilder = PrometheusUtil.getQueryBuilder(QueryBuilderType.RangeQuery);

        //1.get cluster's read & write ops
        URI clusterWriteOpsUri = rangeQueryBuilder.build(PromQL.CLUSTER_WRITE_OPS, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> clusterWriteOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(clusterWriteOpsUri, String.class));

        URI clusterReadOpsUri = rangeQueryBuilder.build(PromQL.CLUSTER_READ_OPS, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> clusterReadOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(clusterReadOpsUri, String.class));
        //construct ops
        Ops ops = new Ops(clusterReadOps.getQueryResult(), clusterWriteOps.getQueryResult());

        //2.get cluster's pg status
        URI totalPgUri = rangeQueryBuilder.build(PromQL.CLUSTER_TOTAL_PG, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> totalPg = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(totalPgUri, String.class));

        URI activePgUri = rangeQueryBuilder.build(PromQL.CLUSTER_ACTIVE_PG, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> activePg = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(activePgUri, String.class));

        URI inActivePgUri = rangeQueryBuilder.build(PromQL.CLUSTER_INACTIVE_PG, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> inActivePg = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(inActivePgUri, String.class));

        URI undersizedPgUri = rangeQueryBuilder.build(PromQL.CLUSTER_UNDERSIZED_PG, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> undersizedPg = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(undersizedPgUri, String.class));

        URI degradedPgUri = rangeQueryBuilder.build(PromQL.CLUSTER_DEGRADED_PG, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> degradedPg = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(degradedPgUri, String.class));

        URI inconsistentPgUri = rangeQueryBuilder.build(PromQL.CLUSTER_INCONSISTENT_PG, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> inconsistentPg = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(inconsistentPgUri, String.class));

        URI downPgUri = rangeQueryBuilder.build(PromQL.CLUSTER_DOWN_PG, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> downPg = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(downPgUri, String.class));

        URI unknownPgUri = rangeQueryBuilder.build(PromQL.CLUSTER_UNKNOWN_PG, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> unknownPg = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(unknownPgUri, String.class));

        double[][] totalPgResult = totalPg.getQueryResult();

        //construct pg
        ClusterMonitorDTO.PG pg = new ClusterMonitorDTO.PG(
            totalPgResult,
            fillResult(totalPgResult, activePg.getQueryResult()),
            fillResult(totalPgResult, inActivePg.getQueryResult()),
            fillResult(totalPgResult, undersizedPg.getQueryResult()),
            fillResult(totalPgResult, degradedPg.getQueryResult()),
            fillResult(totalPgResult, inconsistentPg.getQueryResult()),
            fillResult(totalPgResult, downPg.getQueryResult()),
            fillResult(totalPgResult, unknownPg.getQueryResult()));

        //3.get pg recover rate
        URI recoveryOpsUri = rangeQueryBuilder.build(PromQL.CLUSTER_RECOVERY_OPS, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> recoveryOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(recoveryOpsUri, String.class));

        //construct return object
        ClusterMonitorDTO clusterMonitorDTO = new ClusterMonitorDTO(ops, pg, recoveryOps.getQueryResult());

        return Result.OK(clusterMonitorDTO);
    }

    @Override
    public Result<NodeMonitorDTO> getNodePerformanceMon(NodePerformanceVO nodePerformanceVO) {
        String nodeName = nodePerformanceVO.getNodeName();
        long endTimeStamp = nodePerformanceVO.getDurationVO().getEndTimeStamp();
        long startTimeStamp = nodePerformanceVO.getDurationVO().getStartTimeStamp();
        String step = nodePerformanceVO.getDurationVO().getStep();

        //1.get cpu usage
        RangeQueryBuilder rangeQueryBuilder = PrometheusUtil.getQueryBuilder(QueryBuilderType.RangeQuery);
        URI cpuUsageUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.NODE_CPU_USAGE, new String[]{nodeName, nodeName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> cpuUsage = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(cpuUsageUri, String.class));
        Map<String, double[][]> cpu = cpuUsage.getQueryResultAll(MetricConst.MODE);

        //2.get total and used memory
        URI totalMemoryUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.NODE_MEMORY_TOTAL, new String[]{nodeName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> totalMemory = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(totalMemoryUri, String.class));

        URI usedMemoryUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.NODE_MEMORY_USED, new String[]{nodeName, nodeName, nodeName, nodeName, nodeName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> usedMemory = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(usedMemoryUri, String.class));
        NodeMonitorDTO.MemoryDTO memory = new NodeMonitorDTO.MemoryDTO(totalMemory.getQueryResult(), usedMemory.getQueryResult());

        //3.get network monitor
        //3.1 get network load
        URI receiveLoadUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.NODE_NETWORK_RECEIVE_LOAD, new String[]{nodeName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> receiveLoad = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(receiveLoadUri, String.class));

        URI transmitLoadUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.NODE_NETWORK_TRANSMIT_LOAD, new String[]{nodeName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> transmitLoad = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(transmitLoadUri, String.class));

        //3.2 get drop rate
        URI receiveDropRateUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.NODE_NETWORK_RECEIVE_DROP_RATE, new String[]{nodeName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> receiveDropRate = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(receiveDropRateUri, String.class));

        URI transmitDropRateUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.NODE_NETWORK_TRANSMIT_DROP_RATE, new String[]{nodeName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> transmitDropRate = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(transmitDropRateUri, String.class));

        //3.3 get error rate
        URI receiveErrorRateUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.NODE_NETWORK_RECEIVE_ERROR_RATE, new String[]{nodeName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> receiveErrorRate = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(receiveErrorRateUri, String.class));

        URI transmitErrorRateUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.NODE_NETWORK_TRANSMIT_ERROR_RATE, new String[]{nodeName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> transmitErrorRate = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(transmitErrorRateUri, String.class));

        //3.4 get retransmit rate
        URI retransRateUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.NODE_NETWORK_RETRANS_RATE, new String[]{nodeName, nodeName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> retransRate = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(retransRateUri, String.class));

        NodeMonitorDTO.NetWorkDto netWork = new NodeMonitorDTO.NetWorkDto(
                new NodeMonitorDTO.NetWorkDto.DataDTO(receiveLoad.getQueryResultAll(MetricConst.DEVICE), transmitLoad.getQueryResultAll(MetricConst.DEVICE)),
                new NodeMonitorDTO.NetWorkDto.DataDTO(receiveDropRate.getQueryResultAll(MetricConst.DEVICE), transmitDropRate.getQueryResultAll(MetricConst.DEVICE)),
                new NodeMonitorDTO.NetWorkDto.DataDTO(receiveErrorRate.getQueryResultAll(MetricConst.DEVICE), transmitErrorRate.getQueryResultAll(MetricConst.DEVICE)),
                retransRate.getQueryResult()
        );

        NodeMonitorDTO nodeMonitorDTO = new NodeMonitorDTO(cpu, memory, netWork);

        return Result.OK(nodeMonitorDTO);
    }

    @Override
    public Result<StoragePoolMonitorDTO> getStoragePoolPerformanceMon(StoragePoolPerformanceVO storagePoolPerformanceVO) {
        String poolName = storagePoolPerformanceVO.getPoolName();
        long endTimeStamp = storagePoolPerformanceVO.getDurationVO().getEndTimeStamp();
        long startTimeStamp = storagePoolPerformanceVO.getDurationVO().getStartTimeStamp();
        String step = storagePoolPerformanceVO.getDurationVO().getStep();

        //1.get ops
        RangeQueryBuilder rangeQueryBuilder = PrometheusUtil.getQueryBuilder(QueryBuilderType.RangeQuery);
        URI readOpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.POOL_READ_OPS, new String[]{poolName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> readOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(readOpsUri, String.class));

        URI writeOpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.POOL_WRITE_OPS, new String[]{poolName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> writeOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(writeOpsUri, String.class));
        Ops ops = new Ops(readOps.getQueryResult(), writeOps.getQueryResult());

        //2.get bps
        URI readBpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.POOL_READ_BPS, new String[]{poolName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> readBps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(readBpsUri, String.class));

        URI writeBpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.POOL_WRITE_BPS, new String[]{poolName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> writeBps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(writeBpsUri, String.class));
        Bps bps = new Bps(readBps.getQueryResult(), writeBps.getQueryResult());

        StoragePoolMonitorDTO storagePoolMonitorDTO = new StoragePoolMonitorDTO(ops, bps);

        return Result.OK(storagePoolMonitorDTO);
    }

    @Override
    public Result<Boolean> enableRbdMon(EnableRbdMonitorVO enableRbdMonitorVO) {
        try {
            //check if this pool has enabled
            List<String> enabledPoolsList = enabledPools();
            if (enabledPoolsList.contains(enableRbdMonitorVO.getPoolName())) {
                throw new DsmsEngineException(ResultCode.MON_POOLALLREADEENABLED_ERROR);
            }

            //splicing strings required by ceph
            String enabledPools;
            if (enabledPoolsList.isEmpty()) {
                enabledPools = enableRbdMonitorVO.getPoolName();
            } else {
                enabledPoolsList.add(enableRbdMonitorVO.getPoolName());
                enabledPools = String.join(",", enabledPoolsList);
            }
            RemoteResponse addResponse = rbdApi.enableRbdMon(RemoteCallUtil.generateRemoteRequest(), enabledPools);
            if (Objects.equals(addResponse.getState(), RemoteResponseStatusEnum.SUCCESS.getMessage())) {
                return Result.OK(true);
            } else {
                throw new DsmsEngineException(ResultCode.MON_OPENRBDMON_ERROR);
            }
        } catch (Throwable e) {
            log.error("enable rbd monitoring error:{}", e.getMessage(), e);
            throw DsmsEngineException.exceptionWithMessage(e.getMessage(), ResultCode.MON_OPENRBDMON_ERROR);
        }
    }

    @Override
    public Result<RbdMonitorDTO> getRbdMon(RbdMonitorVO rbdMonitorVO) {
        String poolName = rbdMonitorVO.getPoolName();
        if (!enabledPools().contains(poolName)) {
            throw new DsmsEngineException(ResultCode.MON_RBDMONITORDISABLED_ERROR);
        }

        String rbdName = rbdMonitorVO.getRbdName();
        long endTimeStamp = rbdMonitorVO.getDurationVO().getEndTimeStamp();
        long startTimeStamp = rbdMonitorVO.getDurationVO().getStartTimeStamp();
        String step = rbdMonitorVO.getDurationVO().getStep();

        //1.get ops
        RangeQueryBuilder rangeQueryBuilder = PrometheusUtil.getQueryBuilder(QueryBuilderType.RangeQuery);
        URI readOpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.RBD_READ_OPS, new String[]{poolName, rbdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> readOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(readOpsUri, String.class));

        URI writeOpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.RBD_WRITE_OPS, new String[]{poolName, rbdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> writeOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(writeOpsUri, String.class));

        Ops ops = new Ops(readOps.getQueryResult(), writeOps.getQueryResult());

        //2.get bps
        URI readBpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.RBD_READ_BPS, new String[]{poolName, rbdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> readBps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(readBpsUri, String.class));

        URI writeBpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.RBD_WRITE_BPS, new String[]{poolName, rbdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> writeBps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(writeBpsUri, String.class));

        Bps bps = new Bps(readBps.getQueryResult(), writeBps.getQueryResult());

        //3.get latency
        URI readLatencyUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.RBD_READ_LATENCY, new String[]{poolName, rbdName, poolName, rbdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> readLatency = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(readLatencyUri, String.class));

        URI writeLatencyUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.RBD_WRITE_LATENCY, new String[]{poolName, rbdName, poolName, rbdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> writeLatency = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(writeLatencyUri, String.class));

        Latency latency = new Latency(readLatency.getQueryResult(), writeLatency.getQueryResult());

        RbdMonitorDTO rbdMonitorDTO = new RbdMonitorDTO(ops, bps, latency);

        return Result.OK(rbdMonitorDTO);
    }

    @Override
    public Result<CephFSMonitorDTO> getCephFSMon(DurationVO durationVO) {
        long startTimeStamp = durationVO.getStartTimeStamp();
        long endTimeStamp = durationVO.getEndTimeStamp();
        String step = durationVO.getStep();

        List<FileSystem> fileSystemList = fileSystemService.list();
        if (fileSystemList.isEmpty()) {
            return null;
        }
        FileSystem fileSystem = fileSystemList.get(0);
        String dataPoolId = String.valueOf(fileSystem.getDataPoolId());
        RangeQueryBuilder rangeQueryBuilder = PrometheusUtil.getQueryBuilder(QueryBuilderType.RangeQuery);

        //1.get filesystem's read & write ops
        URI fsWriteOpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.FS_WRITE_OPS, new String[]{dataPoolId}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> fsWriteOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(fsWriteOpsUri, String.class));

        URI fsReadOpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.FS_READ_OPS, new String[]{dataPoolId}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> fsReadOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(fsReadOpsUri, String.class));
        //construct ops
        Ops ops = new Ops(fsReadOps.getQueryResult(), fsWriteOps.getQueryResult());

        //2.get filesystem's client request load
        URI fsMetaDataRequestLoadUri = rangeQueryBuilder.build(PromQL.FS_META_DATA_REQUEST_LOAD, startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> fsMetaDataRequestLoad = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(fsMetaDataRequestLoadUri, String.class));

        CephFSMonitorDTO cephFSMonitorDTO = new CephFSMonitorDTO(ops, fsMetaDataRequestLoad.getQueryResult());

        return Result.OK(cephFSMonitorDTO);
    }

    @Override
    public Result<OsdMonitorDTO> getOsdMon(OsdMonitorVO osdMonitorVO) {
        long startTimeStamp = osdMonitorVO.getDurationVO().getStartTimeStamp();
        long endTimeStamp = osdMonitorVO.getDurationVO().getEndTimeStamp();
        String step = osdMonitorVO.getDurationVO().getStep();
        String osdName = osdMonitorVO.getOsdName();

        //1.get osd ops
        RangeQueryBuilder rangeQueryBuilder = PrometheusUtil.getQueryBuilder(QueryBuilderType.RangeQuery);
        URI osdReadOpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.OSD_READ_OPS, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> osdReadOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(osdReadOpsUri, String.class));

        URI osdWriteOpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.OSD_WRITE_OPS, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> osdWriteOps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(osdWriteOpsUri, String.class));
        Ops osdOps = new Ops(osdReadOps.getQueryResult(), osdWriteOps.getQueryResult());

        //2.get osd bps
        URI osdReadBpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.OSD_READ_BPS, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> osdReadBps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(osdReadBpsUri, String.class));

        URI osdWriteBpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.OSD_WRITE_BPS, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> osdWriteBps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(osdWriteBpsUri, String.class));
        Bps osdBps = new Bps(osdReadBps.getQueryResult(), osdWriteBps.getQueryResult());

        //3.get osd latency
        URI osdApplyLatencyUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.OSD_APPLY_LATENCY, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> osdApplyLatency = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(osdApplyLatencyUri, String.class));

        URI osdCommitLatencyUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.OSD_COMMIT_LATENCY, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> osdCommitLatency = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(osdCommitLatencyUri, String.class));
        Latency osdLatency = new Latency(osdApplyLatency.getQueryResult(), osdCommitLatency.getQueryResult());

        OsdMonitorDTO.OsdData osdData = new OsdMonitorDTO.OsdData(osdOps, osdBps, osdLatency);

        //4.get device latency
        URI deviceReadIopsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.DEVICE_READ_IOPS, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> deviceReadIops = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(deviceReadIopsUri, String.class));

        URI deviceWriteIopsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.DEVICE_WRITE_IOPS, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> deviceWriteIops = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(deviceWriteIopsUri, String.class));
        Iops deviceIops = new Iops(deviceReadIops.getQueryResult(), deviceWriteIops.getQueryResult());

        //5.get device bps
        URI deviceReadBpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.DEVICE_READ_BPS, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> deviceReadBps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(deviceReadBpsUri, String.class));

        URI deviceWriteBpsUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.DEVICE_WRITE_BPS, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> deviceWriteBps = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(deviceWriteBpsUri, String.class));
        Bps deviceBps = new Bps(deviceReadBps.getQueryResult(), deviceWriteBps.getQueryResult());

        //6.get device utilization
        URI deviceUtilizationUri = rangeQueryBuilder.build(PrometheusUtils.getFormatPromQL(PromQL.DEVICE_UTILIZATION, new String[]{osdName}), startTimeStamp, endTimeStamp, step);
        DefaultQueryResult<MatrixData> deviceUtilization = PrometheusUtils.convertQueryResultString(PrometheusUtil.generatePrometheusTemplate().getForObject(deviceUtilizationUri, String.class));

        OsdMonitorDTO.DeviceData deviceData = new OsdMonitorDTO.DeviceData(deviceIops, deviceBps, deviceUtilization.getQueryResult());

        OsdMonitorDTO osdMonitorDTO = new OsdMonitorDTO(osdData, deviceData);
        return Result.OK(osdMonitorDTO);
    }

    private List<String> enabledPools() {
        List<String> enabledPools = new ArrayList<>();
        RemoteResponse getResponse;
        try {
            getResponse = rbdApi.getEnabledRbdMonPoolRequest(RemoteCallUtil.generateRemoteRequest());
            if (Objects.equals(getResponse.getState(), RemoteResponseStatusEnum.SUCCESS.getMessage())) {
                List<FinishedDetail> finished = getResponse.getFinished();
                String outb = finished.get(0).getOutb().trim();
                String[] pools = outb.split(",");
                if (pools.length == 1 && ObjectUtil.isEmpty(pools[0])) {
                    return enabledPools;
                }
                enabledPools = new ArrayList<>(Arrays.asList(pools));
            }
        } catch (Throwable e) {
            throw new DsmsEngineException(ResultCode.MON_GETENABLEDPOOLERROR_ERROR);
        }
        return enabledPools;
    }

    public static double[][] fillResult(double[][] source, double[][] target) {
        if (source == null || target == null) {
            return null;
        }
        double[][] fillResult = new double[source.length][2];

        for (int i = 0; i < source.length; i++) {
            fillResult[i][0] = source[i][0];
        }

        Map<Double, Double> targetMap = new HashMap<>(target.length);
        for (double[] doubles : target) {
            targetMap.put(doubles[0], doubles[1]);
        }

        for (int i = 0; i < fillResult.length; i++) {
            double key = fillResult[i][0];
            fillResult[i][1] = targetMap.getOrDefault(key, 0.0);
        }

        return fillResult;
    }

}
